package com.xsz.pda.net.mq

import android.content.Context
import android.util.Log
import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.TypeReference
import com.blankj.utilcode.util.ThreadUtils
import com.xsz.pda.AppExecutors
import com.xsz.pda.Config
import com.xsz.pda.api.beans.ApiCheckRecord
import com.xsz.pda.event.EmqConn
import com.xsz.pda.net.Constant
import com.xsz.pda.utils.RandomUtil
import info.mqtt.android.service.MqttAndroidClient
import org.eclipse.paho.client.mqttv3.*
import org.greenrobot.eventbus.EventBus
import java.nio.charset.Charset
import java.util.concurrent.ConcurrentHashMap

/**
 * mid: 机器号
 */
class MyMqttClient(private val context: Context,
                   private val androidId: String,
                   projId: String?){

    companion object{ const val TAG = "MyMqttClient" }

    private var mqttClient = MqttAndroidClient(context, "", "")


    private val username = "android"
    private val password = "qwer1234"
    private val myTopic = "device/$androidId"
    private val projectTopic = "project/$projId"
    private val checkBaseTopic = "check/baseConfig"

    private val msgListenerMap: MutableMap<Int, OnMsgListener> = ConcurrentHashMap()


    fun connectEmq() {
        val brokerUrl = "tcp://${Config.ip}:1883"
        // 在 androidId后面加上4位随机字符，以保证不出现clientId冲突 -> 导致一直重连
        val clientId = "$androidId-${RandomUtil.randomString(4)}"
        mqttClient = MqttAndroidClient(context, brokerUrl, clientId)
//        mqttClient.registerResources()
        mqttClient.setCallback(object : MqttCallbackExtended {
            override fun messageArrived(topic: String?, message: MqttMessage?) {
               AppExecutors.diskIO.execute {
                   message?.let {
                       topic?.let {
                           //设备控制消息
                           val jsonString = String(message.payload, Charset.forName("UTF-8"))
                           val info = JSON.parseObject<Map<String, Any>>(jsonString, object : TypeReference<Map<String?, Any?>?>() {}.type)
                           val what = info["what"] as Int
                           val data = JSON.toJSONString(info["data"])
                           msgListenerMap[what]?.onMessage(data)
                           Log.d(TAG, "receive: {what:$what,data: ${data.subSequence(0, 255)}")
                       }

                   }
               }
            }

            override fun connectionLost(cause: Throwable?) {
                EventBus.getDefault().post(EmqConn(false))
                Log.i(TAG, "连接断开: ${cause.toString()}")
            }


            override fun deliveryComplete(token: IMqttDeliveryToken) {
//                Log.i(TAG, "DeliveryComplete ${token.message}")
            }

            override fun connectComplete(reconnect: Boolean, serverURI: String?) {
                Log.i(TAG, "连接 broker 成功")
                msgListenerMap[Constant.CONNECT_SUCCESS]?.onMessage("")
                subscribe(topic = myTopic, qos = 2)
                subscribe(topic = projectTopic, qos = 2)
                subscribe(topic = checkBaseTopic, qos = 0)
            }
        })
        doConnect()
    }

    private fun doConnect(){
        val options = MqttConnectOptions()
        options.isAutomaticReconnect = true
        options.isCleanSession = true
        options.userName = username
        options.password = password.toCharArray()
        options.connectionTimeout = 5
        options.keepAliveInterval = 10
        try {
            mqttClient.connect(options, null, object : IMqttActionListener {
                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    Log.i(TAG, "连接 broker 成功")
                    EventBus.getDefault().post(EmqConn(true))
                    subscribe(topic = myTopic, qos = 2)
                    subscribe(topic = projectTopic, qos = 2)
                }

                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                    Log.e(TAG, "连接broker失败, errorMsg = ${exception?.message}",exception)
                    EventBus.getDefault().post(EmqConn(false))
                    exception?.message?.let {
                        if (it == "已连接客户机"){ return@onFailure }
                        else { reConnect() }
                    }
                }
            })
        } catch (e: MqttException) {
            Log.e(TAG, "连接broker异常, errorMsg = ${e.message}")
            EventBus.getDefault().post(EmqConn(false))
            reConnect()
        }
    }

    private fun reConnect() {
        mqttClient.unregisterResources()
        ThreadUtils.runOnUiThreadDelayed({ connectEmq() },2000)
    }


    private fun subscribe(topic: String, qos: Int = 1) {
        try {
            mqttClient.subscribe(topic, qos, null, object : IMqttActionListener {
                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    EventBus.getDefault().post(EmqConn(true))
                    Log.i(TAG, "订阅主题： $topic 成功")
                }
                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                    Log.e(TAG, "订阅主题： $topic 失败 errMsg = ${exception?.message}")
                    //TODO 记录订阅失败日志
                }
            })
        } catch (e: MqttException) {
            Log.e(TAG, "订阅主题： $topic 异常 errMsg = ${e.message}")
            //TODO 记录订阅失败日志
        }
    }

    fun unsubscribe(topic: String) {
        mqttClient.unsubscribe(topic, null, object : IMqttActionListener {
            override fun onSuccess(asyncActionToken: IMqttToken?) {
                Log.i(TAG, "Unsubscribed to $topic")
            }
            override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                Log.e(TAG, "Failed to unsubscribe $topic")
            }
        })
    }

    fun setMsgListener(what: Int, msgListener: OnMsgListener) {
        msgListenerMap[what] = msgListener
    }


    private fun publish(topic: String, msg: String, qos: Int = 1, retained: Boolean = false) {
        try {
            val message = MqttMessage()
            message.payload = msg.toByteArray()
            message.qos = qos
            message.isRetained = retained
            if (!mqttClient.isConnected) { return }
            mqttClient.publish(topic, message, null, object : IMqttActionListener {
                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    Log.d(TAG, "Success to publish $message to $topic")
                }
                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                    Log.d(TAG, "Failed to publish $msg to $topic")

                }
            })
        } catch (e: MqttException) {
            Log.e(TAG, "Publish Error", e)
        }
    }


    fun syncItem(syncTicketItem: ApiCheckRecord){
        val map: MutableMap<String, Any> = HashMap()
        map["what"] = Constant.SYNC_CHECK_RECORD
        map["data"] = syncTicketItem
        publish(projectTopic, JSON.toJSONString(map), 0, false)
    }

    fun release() {
        mqttClient.disconnect(null, object : IMqttActionListener {
            override fun onSuccess(asyncActionToken: IMqttToken?) {
                Log.d(TAG, "Disconnected")
            }
            override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                Log.d(TAG, "Failed to disconnect")
            }
        })
        mqttClient.unregisterResources()
    }

    fun isOnline() = mqttClient.isConnected

}